package com.kobeliu.controller;


import com.kobeliu.enums.AggsUserType;
import com.kobeliu.enums.Sex;
import com.kobeliu.pojo.eo.AggsUserEO;
import com.kobeliu.pojo.eo.Stu;
import com.kobeliu.pojo.vo.MapVO;
import com.kobeliu.result.GraceJSONResult;
import com.kobeliu.result.ResponseStatusEnum;
import com.kobeliu.utils.LocalDateUtils;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.aggregations.Aggregation;
import org.elasticsearch.search.aggregations.AggregationBuilder;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.Aggregations;
import org.elasticsearch.search.aggregations.bucket.histogram.*;
import org.elasticsearch.search.aggregations.bucket.terms.Terms;
import org.elasticsearch.search.aggregations.bucket.terms.TermsAggregationBuilder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.convert.ThreeTenBackPortConverters;
import org.springframework.data.elasticsearch.core.AggregationsContainer;
import org.springframework.data.elasticsearch.core.ElasticsearchRestTemplate;
import org.springframework.data.elasticsearch.core.SearchHits;
import org.springframework.data.elasticsearch.core.query.NativeSearchQuery;
import org.springframework.data.elasticsearch.core.query.NativeSearchQueryBuilder;
import org.springframework.data.elasticsearch.core.query.Query;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.lang.annotation.Native;
import java.time.LocalDateTime;
import java.time.temporal.ChronoUnit;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.stream.Stream;

@RestController
@RequestMapping("aggsUser")
@Slf4j
public class AggsUserController {

    @Autowired
    private ElasticsearchRestTemplate esTemplate;

    @Autowired
    private ThreadPoolExecutor threadPoolExecutor;

    @PostMapping("createUser")
    public GraceJSONResult createUser(){

        AggsUserEO eo1 = new AggsUserEO();
        eo1.setType(AggsUserType.USER.type);
        eo1.setUserId("10023");
        eo1.setFace("");
        eo1.setNickname("kobeLiu");
        eo1.setIndustry("");
        eo1.setSex(Sex.man.type);
        eo1.setProvince("江苏");
        eo1.setCompanyName("");
        eo1.setCompanyId("");
        eo1.setCreateDate(LocalDateUtils.format(LocalDateTime.now(),LocalDateUtils.DATE_PATTERN));

        esTemplate.save(eo1);
        return GraceJSONResult.ok();
    }

    /**
     * 地图 统计每个地区各有多少人在图表上显示 aggs_user.province
     * @return
     */
    @GetMapping("/getUserCountsMap")
    public GraceJSONResult userCountsMap(){


        //聚合搜索，根据province进行分类
        TermsAggregationBuilder aggregationBuilder = AggregationBuilders
                .terms("map_count")
                .field("province");

        Query query = new NativeSearchQueryBuilder()
                .withAggregations(aggregationBuilder)
                .build();

        SearchHits<AggsUserEO> search = esTemplate.search(query, AggsUserEO.class);

        Aggregations aggregations = (Aggregations)search.getAggregations().aggregations();

        Terms mapCount = aggregations.get("map_count");
        List<? extends Terms.Bucket> buckets = mapCount.getBuckets();
        List<MapVO> vos = new ArrayList<>();

        buckets.stream().forEach(obj -> {
            String keyAsString = obj.getKeyAsString();
            MapVO pending = new MapVO();
            pending.setName("".equals(keyAsString)?"台湾":keyAsString);
            pending.setValue(obj.getDocCount());
            vos.add(pending);
        });

        return GraceJSONResult.ok(vos);
    }

    /**
     * 根据aggs_user.type 进行分类，统计每种type个有多少人
     * @return
     */
    @GetMapping("/getUserCountsTotal")
    public GraceJSONResult getUserCountsTotal(){

        TermsAggregationBuilder aggregationBuilder = AggregationBuilders
                .terms("daily_counts")
                .field("type");

        Query query = new NativeSearchQueryBuilder()
                .withAggregations(aggregationBuilder)
                .build();

        SearchHits<AggsUserEO> search = esTemplate.search(query, AggsUserEO.class);

        //拿到下面的结果集
        Aggregations aggregations = (Aggregations)search.getAggregations().aggregations();
        //
        Terms dailyCounts = aggregations.get("daily_counts");

        //解析
        List<? extends Terms.Bucket> buckets = dailyCounts.getBuckets();

        Map<String,Long> countsMap = new HashMap<>();

        buckets.stream().forEach( res -> {
            //每一项的type
            String keyAsString = res.getKeyAsString();
            //对应的值
            long docCount = res.getDocCount();

            countsMap.put(keyAsString,docCount);
        });

        return GraceJSONResult.ok(countsMap);
    }


    /**
     *
     * @return
     */
    @GetMapping("/getUserCountsDaily")
    public GraceJSONResult getUserCountsDaily(){
        //从今天开始，往前推七天
        //每天的日期，以及每天对应的四种type的人数 1，2，3，4

        Map<String,Object> lineMap = new HashMap<>();
        //改并行
        CompletableFuture<Void> userFuture = CompletableFuture.runAsync(()-> lineMap
                .put("newUsers",getDailyByUserType(AggsUserType.USER))
                ,threadPoolExecutor
        );

        CompletableFuture<Void> hrFuture = CompletableFuture.runAsync(()-> lineMap
                        .put("newHRs",getDailyByUserType(AggsUserType.HR))
                ,threadPoolExecutor
        );

        CompletableFuture<Void> companyFuture = CompletableFuture.runAsync(()-> lineMap
                        .put("newCompanies",getDailyByUserType(AggsUserType.COMPANY))
                ,threadPoolExecutor
        );

        CompletableFuture<Void> candFuture = CompletableFuture.runAsync(()-> lineMap
                        .put("successCandidates",getDailyByUserType(AggsUserType.ENTRY))
                ,threadPoolExecutor
        );

        CompletableFuture<Void> dayFuture = CompletableFuture.runAsync(()-> lineMap
                        .put("days",getSevenDays())
                ,threadPoolExecutor
        );

        try {
            CompletableFuture.allOf(userFuture,hrFuture,candFuture,companyFuture,dayFuture).get();
        } catch (Exception e) {
            return GraceJSONResult.errorCustom(ResponseStatusEnum.USER_STATUS_ERROR);
        }
//        lineMap.put("newUsers",getDailyByUserType(AggsUserType.USER));
//        lineMap.put("newHRs",getDailyByUserType(AggsUserType.HR));
//        lineMap.put("newCompanies",getDailyByUserType(AggsUserType.COMPANY));
//        lineMap.put("successCandidates",getDailyByUserType(AggsUserType.ENTRY));
//        lineMap.put("days",getSevenDays());

        return GraceJSONResult.ok(lineMap);

    }

    private List<String> getSevenDays(){
        List<String> dayList = new ArrayList<>();
//        log.info("================================");

        for (int i = 6; i > -1; i--) {
            String day = LocalDateUtils
                    .format(LocalDateUtils.
                            minus(LocalDateTime.now(),i,ChronoUnit.DAYS),
                            LocalDateUtils.DATE_PATTERN);
//            log.info("day:"+day);
            dayList.add(day);
        }
//        log.info("================================");

        return dayList;
    }

    /**
     * 根据 type 和日期区间来查询 每一天该type的统计数据
     * @param type
     * @return
     */
    private List<Long> getDailyByUserType(AggsUserType type){

        // 根据 type 和日期区间来查询 每一天该type的统计数据
        //聚合构造器
        //定位今天和一周前的时间
        String today = LocalDateUtils.format(LocalDateTime.now(),LocalDateUtils.DATE_PATTERN);
        String beforeWeek = LocalDateUtils.format(LocalDateUtils
                .minus(LocalDateTime.now(),
                        6,
                        ChronoUnit.DAYS),LocalDateUtils.DATE_PATTERN);

        DateHistogramAggregationBuilder builder = AggregationBuilders
                .dateHistogram("date_counts")
                .minDocCount(0)
                .field("createDate")
                .calendarInterval(DateHistogramInterval.DAY)
                .format("yyyy-MM-dd")
//                .offset("-8h")
                .extendedBounds(new LongBounds(beforeWeek,today));

        //bool查询
        BoolQueryBuilder queryBuilder = QueryBuilders.boolQuery();
        queryBuilder.must(QueryBuilders.termQuery("type",type.type));

        //区间查询
        queryBuilder.must(QueryBuilders
                .rangeQuery("createDate")
                .gte(beforeWeek)
                .lte(today)
        );

        NativeSearchQuery nativeSearchQuery = new NativeSearchQueryBuilder()
                .withQuery(queryBuilder)
                .withAggregations(builder)
                .build();

        SearchHits<AggsUserEO> search = esTemplate.search(nativeSearchQuery, AggsUserEO.class);

        Aggregations aggregations = (Aggregations)search.getAggregations().aggregations();
        ParsedDateHistogram dateCounts = aggregations.get("date_counts");
        List<? extends Histogram.Bucket> buckets = dateCounts.getBuckets();
        List<Long> longList = new ArrayList<>();

        buckets.stream().forEach(obj ->{
            Long count = obj.getDocCount();
            longList.add(count);
        });
        return longList;


    }

}
